#pragma once
#include "../mqcommon/helper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/msg.pb.h"
#include <unordered_map>
#include <mutex>
#include <list>

namespace wzf
{
    using MessagePtr = std::shared_ptr<Message>;            //这个和protobuf里的重复了
}
class MessageMapper
{
    #define DATEFILE_SUBFIX ".mqd";
    #define TMPFILE_SUBFIX ".mqd.tmp";
public:
    MessageMapper(std::string& basedir, const std::string& mqName)
        :_msgQueueName(mqName)
    {
        if(basedir.back() != '/') basedir += '/';       //难怪，chuan  ccc 不走/对退出
        _dataFile = basedir + mqName + DATEFILE_SUBFIX;
        _tmpFile = basedir + mqName + TMPFILE_SUBFIX;
        if(FileHelper::exists(basedir) == false)
            assert(FileHelper::createDirectory(basedir));       //目录创建失败直接中断

        createMsgQueueFile();  //果然是这里

    }
    bool createMsgQueueFile()
    {
        if(FileHelper::exists(_dataFile) == false)      //多一步判断
            assert(FileHelper::createFile(_dataFile));
        if(FileHelper::exists(_tmpFile) == false)
            assert(FileHelper::createFile(_tmpFile));
        return true;
    }
    void removeMsgQueueFile()
    {
        FileHelper::removeFile(_dataFile);
        FileHelper::removeFile(_tmpFile);
    }

    bool insert(wzf::MessagePtr ptr)
    {
        return insert(ptr, _dataFile);
    }
    bool insert(wzf::MessagePtr ptr, const std::string& file)
    {
        std::string payLode = ptr->paylode().SerializeAsString(); //序列化有效载荷
        wzf::MessagePtr ptr1 = std::make_shared<Message>();          //这样竟然也不行
        // ptr1->mutable_paylode()->ParseFromString(payLode);
        // DLOG("%s", ptr1->paylode().property().id().c_str());     //不对这里
        // DLOG("%s", ptr1->paylode().body().c_str());
        FileHelper helper(file);
        size_t fileSize = helper.size();
        uint32_t payLodeSize = payLode.size();
        // DLOG("%d", payLodeSize);
        // DLOG("%s", payLode.c_str()); 

        if(helper.write(reinterpret_cast<char*>(&payLodeSize), fileSize, 4) == false) //填充4字节
        {
            DLOG("描述有效载荷长度的4字节字段, 写入失败");
            return false;
        }
        if(helper.write(payLode, fileSize + 4, payLodeSize) == false)     //填充序列化的有效载荷
        {
            DLOG("有效载荷, 写入失败");
            return false;
        }
        // DLOG("filesize: %ld", fileSize);
        ptr->set_offset(fileSize + 4);
        ptr->set_len(payLode.size());       //有效载荷的长度
        return true;
    }
    bool remove(wzf::MessagePtr ptr)             //为什么要重新写入：因为序列化后是二进制的
    {
        ptr->mutable_paylode()->set_valid("0");             //paylode()返回的是const的
        // DLOG("%s", ptr->paylode().valid().c_str());
        std::string newPayLode = ptr->paylode().SerializeAsString();
        if(newPayLode.size() != ptr->len())
        {
            DLOG("序列化后长度不一致，不允许写入");
            return false;
        }

        FileHelper helper(_dataFile);
        if(helper.write(newPayLode, ptr->offset(), newPayLode.size()) == false) //这里不能加4，上面加4了
        {
            DLOG("置有效位为0时,重新写入是失败");
            return false;
        }
        return true;
    }
    std::list<wzf::MessagePtr> gc()
    {
        std::list<wzf::MessagePtr> result;
        FileHelper helperDataFile(_dataFile);
        FileHelper helperTmpFile(_tmpFile);
        uint32_t offset = 0, payLodeSize = 0;
        // DLOG("_dataFilesize:%ld", helperDataFile.size());

        while(offset < helperDataFile.size())
        {
            if(!helperDataFile.read(reinterpret_cast<char*>(&payLodeSize), offset, 4))//读取序列化后的有效载荷长度
            {
                DLOG("前四个字节，读取失败");
            }
            offset += 4;
            // DLOG("%d", payLodeSize); 
            std::string payLode(payLodeSize, 0);
            if(!helperDataFile.read(payLode, offset, payLodeSize))//读取序列化后的有效载荷
            {
                DLOG("有效载荷，读取失败");
            }
            // DLOG("%s", payLode.c_str()); 

            offset += payLodeSize;                    //漏了一次
            wzf::MessagePtr ptr = std::make_shared<Message>();
            ptr->mutable_paylode()->ParseFromString(payLode); ///难怪反序列化出不来
                    
            

            if(ptr->paylode().valid() == "0") 
            {
                DLOG("发现无效消息：%s:%s", ptr->paylode().property().id().c_str(),ptr->paylode().body().c_str());
                continue;             //发现哪里不对了，把一个没有序列化的存进去了
            }
            else DLOG("发现有效消息：%s:%s", ptr->paylode().property().id().c_str(),ptr->paylode().body().c_str());
            if(insert(ptr, _tmpFile) == false)          //写入数据
            {
                DLOG("向临时文件, 写入失败");
            }
            // DLOG("%s", ptr->paylode().property().id().c_str());
            // DLOG("%s", ptr->paylode().body().c_str());
            // DLOG("临时文件大小%ld", helperTmpFile.size());
            result.push_back(ptr);          //保存有效Message
        }

        FileHelper::removeFile(_dataFile);
        helperTmpFile.rename(_dataFile);
        FileHelper::createFile(_tmpFile);
        return result;
    }
private:
    std::string _msgQueueName;
    std::string _dataFile;
    std::string _tmpFile;
};

class QueueMessage
{
public:
    using Ptr = std::shared_ptr<QueueMessage>;
    QueueMessage(std::string basedir, const std::string& queueName)
        :_qname(queueName), _valid_count(0), _total_count(0), _mapper(basedir, queueName)
    {
        // recovery();         //把初始化和恢复隔开，让他初始化快点出来 这里就不能有了，这样就初始化两次了
    }
    void recovery()
    {
        std::unique_lock<std::mutex> lock(_mutex);      //恢复持久化消息
        _msgs = _mapper.gc();
        for(auto& msg : _msgs)
        {
            _durable_msg.insert({msg->paylode().property().id(), msg}); //难道这里没有成功吗
            // DLOG("%s", msg->paylode().body().c_str());                                                                                      //DLOG成功了
        }
        _valid_count = _total_count = _msgs.size();
    }
    bool insert(BasicProperty* bp, const std::string& body, bool queueIsDuarble)
    {
        // DLOG("errno:%s", strerror(errno));
        DLOG("消息body：%s开始创建", body.c_str());
        DeliveryMode mode;
        wzf::MessagePtr pmsg = std::make_shared<Message>();
        if(bp)
        {
            mode = queueIsDuarble ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
            pmsg->mutable_paylode()->mutable_property()->set_id(bp->id());
            pmsg->mutable_paylode()->mutable_property()->set_delivery_mode(mode);
            pmsg->mutable_paylode()->mutable_property()->set_routing_key(bp->routing_key());
        }
        else
        {
            mode = queueIsDuarble ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
            pmsg->mutable_paylode()->mutable_property()->set_id(UUIDHelper::uuid());
            pmsg->mutable_paylode()->mutable_property()->set_delivery_mode(mode);
            pmsg->mutable_paylode()->mutable_property()->set_routing_key("");
        }
        pmsg->mutable_paylode()->set_body(body);
        std::unique_lock<std::mutex> lock(_mutex);
        if(pmsg->paylode().property().delivery_mode() == DeliveryMode::DURABLE)
        {
            pmsg->mutable_paylode()->set_valid("1");
            if(_mapper.insert(pmsg) == false)                               //文件中一份
            {
                DLOG("写入文件失败, body:%s", body.c_str());
                return false;
            }
            _total_count++;
            _valid_count++;
            _durable_msg.insert({pmsg->paylode().property().id(), pmsg}); //内存中保存一份持久化的消息
        }
        else pmsg->mutable_paylode()->set_valid("0");
        DLOG("isempty:_msgs:%d", _msgs.empty());            //////////////怎么会是空的？
        _msgs.push_back(pmsg);
        DLOG("内存中保存的消息的个数_msgs:%ld", _msgs.size());  ////////////////////////////////////////////////////为什么一直是1
        DLOG("_total_count:%ld, _valid_count:%ld", _total_count, _valid_count);                     
        return true;
    }
    wzf::MessagePtr front()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        if(_msgs.empty()) return wzf::MessagePtr();
        wzf::MessagePtr msg = _msgs.front();
        // DLOG("front():%s", msg->paylode().body().c_str());
        _msgs.pop_front();
        _waitack_msg.insert({msg->paylode().property().id(), msg});     //推送出去，放入等待响应hash
        return msg;
    }
    bool remove(const std::string& msgID)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _waitack_msg.find(msgID);
        for(auto &aa:_waitack_msg)
            DLOG("%s", aa.first.c_str());               ////////////////////////////////打印的混乱
        if(it == _waitack_msg.end())
        {
            DLOG("没有该消息msgid:%s", msgID.c_str());
            return true;
        }
        wzf::MessagePtr pmsg = it->second;
        if(pmsg->paylode().property().delivery_mode() == DeliveryMode::DURABLE)   //判断是否是持久化
        {
            _mapper.remove(pmsg);
            _durable_msg.erase(msgID);      
            _valid_count--;
            DLOG("%ld", _valid_count);
            if(gcCheck()) _mapper.gc();
        }
        // _total_count--;
        _waitack_msg.erase(msgID);
        return true;
    }
    bool gcCheck()
    {
        return _total_count > 2000 && (double)_valid_count / _total_count < 0.5;
    }
    void gc()
    {
        if(!gcCheck()) return;
        for(auto& msg : _msgs)  
        {
            auto it = _durable_msg.find(msg->paylode().property().id());

        }
        //我目前觉得这个函数不需要
    }
    size_t msgsCount()       //待推送消息个数
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _msgs.size();
    }
    size_t totalCount()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _total_count;
    }
    size_t validCount()
    {   
        std::unique_lock<std::mutex> lock(_mutex);
        return _valid_count;
    }
    size_t waitackCount()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _waitack_msg.size();
    }
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _mapper.removeMsgQueueFile();
        // DLOG("_mapper.removeMsgQueueFile();");   //没有被调
        // sleep(10);
        _msgs.clear();
        _durable_msg.clear();
        _waitack_msg.clear();
    }

private:
    std::mutex _mutex;              //一个队列一把锁、一个文件一把锁
    std::string _qname;
    std::list<wzf::MessagePtr> _msgs;    //可获取消息，链表管理起来
    size_t _valid_count;        //文件中(持久化的有效的数量)有效消息的个数：等于内存中_durable_msg.size();
    size_t _total_count;        //文件中(持久化的数量)所有消息的个数
    MessageMapper _mapper;          //磁盘中的消息
    std::unordered_map<std::string, wzf::MessagePtr> _durable_msg; //内存中保存一份持久化的消息
    std::unordered_map<std::string, wzf::MessagePtr> _waitack_msg; //等待响应的消息
};

class MessageManager
{
public:
    using Ptr = std::shared_ptr<MessageManager>;

    MessageManager(const std::string& basedir)
        :_basedir(basedir)
    {}
    
    void initQueueMessage(const std::string& qname)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it != _queue_msgs.end())
            {
                DLOG("qname:%s已经存在", qname.c_str());
                return;
            }
            qmsg = std::make_shared<QueueMessage>(_basedir, qname); //找到了，这里出来没有的、、、、、、、、、
            _queue_msgs.insert({qname, qmsg});
        }

        qmsg->recovery();

        // DLOG("qmsg->recovery();");
    }

    void destoryQueueMessage(const std::string& qname)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return;
            }
            qmsg = it->second;
            _queue_msgs.erase(it);
        }
        qmsg->clear();
    }
    
    void insert(const std::string& qname, BasicProperty* bp, const std::string& body, bool queueIsDuarble)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return;
            }
            qmsg = it->second;
        }
        DLOG("有走到这？");
        qmsg->insert(bp, body, queueIsDuarble);
    }
    
    wzf::MessagePtr front(const std::string& qname)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return wzf::MessagePtr();
            }
            qmsg = it->second;
        }
        return qmsg->front();
    }
    
    void ack(const std::string& qname, const std::string& msgID)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return;
            }
            qmsg = it->second;
        }
        qmsg->remove(msgID);
    }
    
    size_t msgsCount(const std::string& qname)       //待推送消息的个数
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return 0;
            }
            qmsg = it->second;
        }
        return qmsg->msgsCount();
    }
    
    size_t totalCount(const std::string& qname)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return 0;
            }
            qmsg = it->second;
        }
        return qmsg->totalCount();
    }
    
    size_t validCount(const std::string& qname)
    {   
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return 0;
            }
            qmsg = it->second;
        }
        return qmsg->validCount();
    }
    
    size_t waitackCount(const std::string& qname)
    {
        QueueMessage::Ptr qmsg;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end()) 
            {
                DLOG("qname:%s不存在", qname.c_str());
                return 0;
            }
            qmsg = it->second;
        }
        return qmsg->waitackCount();
    }
    
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        for(auto &qmsg : _queue_msgs)
        {
            qmsg.second->clear();
        }
    }
private:
    std::mutex _mutex;
    std::string _basedir;
    std::unordered_map<std::string, QueueMessage::Ptr> _queue_msgs;
};

